package com.aispeech.asr.apis.ssc.websocket.connection;

import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
import org.springframework.web.reactive.socket.adapter.ReactorNettyWebSocketSession;

import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.enums.ErrorDescEnum;
import com.aispeech.asr.apis.ssc.websocket.emitter.ClientEmitter;
import com.aispeech.asr.apis.ssc.websocket.emitter.Emitter;
import com.aispeech.asr.apis.ssc.websocket.emitter.EmitterState;
import com.aispeech.asr.apis.ssc.websocket.emitter.EmitterType;
import com.aispeech.asr.comm.data.vo.ResultOnline;
import com.aispeech.asr.comm.exceptions.AsrException;
import com.aispeech.asr.webase.component.BigFileHttpPoster;
import com.aispeech.jarpc.ArpcException;
import com.aispeech.jarpc.JArpc;
import com.aispeech.jarpc.TextBin;
import com.aispeech.jarpc.op.sd.SdResult;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;

import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Slf4j
@Component
public class EmitterManager {
	
	@Autowired
    AppConfig config;
	
	@Autowired
	BigFileHttpPoster bigFileHttpPoster;
	
    private final ArrayBlockingQueue<Message> messages = new ArrayBlockingQueue<>(65536);
    private Map<String, Emitter> clientEmitterMap;
    private volatile boolean running = false;

    @PostConstruct
    private void init() {
        running = true;
        clientEmitterMap = new ConcurrentHashMap<>();

        new Thread(this::loop).start();
        new Thread(this::closeWebsocketSessionLoop).start();
    }

    @PreDestroy
    private void finit() {
        running = false;
    }
    
    public int clientEmitterMapSize() {
    	return clientEmitterMap.size();
    }
    
    public ClientEmitter getClientEmitter(String clientSessionId) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	return clientEmitter;
    }
   
    public boolean isClientRunning(String clientSessionId) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	return clientEmitter.getState() == EmitterState.RUNNING;
    }
    
    public EmitterState getClientState(String clientSessionId) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	return clientEmitter.getState();
    }
	
	/**
     * 后端服务返回后在发送到客户端
     * @param id
     * @param type
     * @param message
     */
	public void immediateSendCustomerClient(String clientSessionId, String message) {
		messages.add(Message.alignment(clientSessionId, message, "immediate")); 
	}
	
	/**
     * 后端服务返回后在发送到客户端
     * @param id
     * @param type
     * @param message
     */
	public void immediateSendCustomerClient(Message message) {
		messages.add(message); 
	}
	
	public void sendBackServer(String emitterId, Message message) {
        messages.add(Message.of(emitterId, message.getMessage(), message.getBuffer()));
    }
    
    public void addTimeToClientReceivedList(String clientSessionId) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	long receiveTime = System.currentTimeMillis();
    	clientEmitter.getReceivedList().add(receiveTime);
    	clientEmitter.setLastReceiveTime(receiveTime);
    }  
    
    public void updateClientState(String clientSessionId, EmitterState state) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);

    	clientEmitter.setState(state);
    	//clientEmitter.setLastReceiveTime(System.currentTimeMillis());
    }

    public void addClientWaitQueue(String clientSessionId, String asrMessage) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	clientEmitter.getReceivedQueue().add(Message.of(clientSessionId, asrMessage, null));
    }

    public void doProcess(String clientSessionId, DataBuffer dataBuffer) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	//clientEmitter.getReceivedQueue().add(Message.of(clientSessionId, null, dataBuffer));
    	
    	if(config.isSaveAudio() && dataBuffer.readableByteCount()>0 ) {
    		NettyDataBuffer writeDataBuffer = DateBufferFactory.getInstance().wrap(dataBuffer.asByteBuffer());
    		clientEmitter.writeReceiveBuffer(writeDataBuffer);
        	writeDataBuffer.release();
    	}
    	
    	clientEmitter.doProcess(dataBuffer);
    }
    
    public void doSpeakerDiarisation(String clientSessionId, TextBin textBin) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	clientEmitter.doSpeakerDiarisation(textBin,0);
    }
    
    public void doResponse(String clientSessionId, SdResult sdResult, int eof) {
    	String result = JSON.toJSONString(ResultOnline.ok(sdResult,eof));
    	messages.add(Message.alignment(clientSessionId, result, eof ==1 ? "end":"")); 
    }
    
    public void doErrorResponse(String clientSessionId, Exception exception) {
    	
    	int code = ErrorDescEnum.ERROR.getCode();
    	String message = ErrorDescEnum.ERROR.getMessage();
    	
    	if(exception instanceof AsrException) {
    		code = ((AsrException)exception).getCode();
    		message = ((AsrException)exception).getMessage();
    	}else if(exception instanceof ArpcException){
    		//code = ((ArpcException)exception).getCode();
    		message = ((ArpcException)exception).getMessage();
    	}
    	
    	String result = JSON.toJSONString(ResultOnline.fail(code,message));
    	messages.add(Message.alignment(clientSessionId, result, "")); 
    }

    public Mono<Void> addSession(WebSocketSession session) {
    	Emitter emitter = null;
    	try {
    		emitter = addEmitter(session,session.getId(), EmitterType.CLIENT);
    	}catch (Error e) {
    		log.info("=======addSession error=========");
    		return Mono.error(e);
		}
        return session.send(Flux.create(emitter));
    }
    
    
    @SneakyThrows
    public Mono<WebSocketCloseStatus> addCloseStatusToWebSession(WebSocketSession session, Consumer<? super WebSocketCloseStatus> onClosed) {
    	log.info("=======addCloseStatusToWebSession start=========");
    	ReactorNettyWebSocketSession reactorNettyWebSocketSession = (ReactorNettyWebSocketSession) session;
        Method method;
		method = AbstractWebSocketSession.class.getDeclaredMethod("getDelegate");
		method.setAccessible(true);
		ReactorNettyWebSocketSession.WebSocketConnection connection;
		connection = (ReactorNettyWebSocketSession.WebSocketConnection) method.invoke(reactorNettyWebSocketSession);
		

		return connection.getInbound().receiveCloseStatus().doOnNext(onClosed);
    }
    

    public Emitter addEmitter(WebSocketSession session, String id, EmitterType type ) {
        Emitter emitter;
        Map<String, Emitter> map;

        emitter = new ClientEmitter(session,id, this);
        map = clientEmitterMap;
        map.put(id, emitter);
        return emitter;
    }

    public void removeClientEmitter(String clientSessionId) {
    	log.info("removeClientEmitter ===========================clientSessionId:{}",clientSessionId);
        ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.remove(clientSessionId);

        if(clientEmitter != null) {
        	if(clientEmitter.getJarpcVad() != null) {
        		clientEmitter.getJarpcVad().close();
        	}
        	if(clientEmitter.getJarpcSd() != null) {
        		clientEmitter.getJarpcSd().close();
        	}
	        clientEmitter.getSession().close(CloseStatus.NORMAL).subscribe();
        }
    }

    public WebSocketMessage textMessage(String payload) {
        byte[] bytes = payload.getBytes(StandardCharsets.UTF_8);
        return new WebSocketMessage(WebSocketMessage.Type.TEXT, DateBufferFactory.getInstance().wrap(bytes));
    }

    public WebSocketMessage binaryMessage(DataBuffer dataBuffer) {
        return new WebSocketMessage(WebSocketMessage.Type.BINARY, dataBuffer);
    }
    
    public void setClientProperties(String clientSessionId, String reuqestQuery, String requestId, JSONObject startObj, JSONObject audioObject, JArpc jarpcVad, JArpc jarpcSd, boolean resetFlag) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);

        if (clientEmitter != null) {
        	clientEmitter.setSessionId(requestId);
        	clientEmitter.setRequestQuery(reuqestQuery);
        	clientEmitter.setStartObj(startObj);
        	clientEmitter.setAudioObj(audioObject);
        	//clientEmitter.setSedOnlineCallback(sedOnlineCallback);
        	clientEmitter.setJarpcVad(jarpcVad);
        	clientEmitter.setJarpcSd(jarpcSd);
        	
        	String audioPath = String.format(
                    "/audios/%s/%s.%s",
                    LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")),
                    requestId,
                    audioObject.getString("audioType"));
			clientEmitter.setAudioPath(audioPath);
        }
    }
    
    public void backupAudio(String clientSessionId) {
    	ClientEmitter clientEmitter = (ClientEmitter) clientEmitterMap.get(clientSessionId);
    	if(config.isSaveAudio()) {
    		log.info("clientEmitter.getReceiveBuffer:{}",clientEmitter.getReceiveBuffer());
    		if(StringUtils.isNotBlank(clientEmitter.getAudioPath()) && clientEmitter.getReceiveBuffer() != null &&  clientEmitter.getReceiveBuffer().readableByteCount() > 0) {
    			backup(clientEmitter.getAudioPath(),clientEmitter.getReceiveBuffer().asInputStream());
    		}
    	}
    }
    
    private void backup(String filePath, InputStream inputStream) {
        //String objectKey = filePath.replace("/backup", "/audios/sentence");

        try {
            byte[] buff = new byte[inputStream.available()];
            inputStream.read(buff);
            bigFileHttpPoster.add(filePath, filePath, new ByteArrayResource(buff));
        } catch (IOException e) {
            log.warn("backup file: {} failed.", filePath);
        }
    }


    private void loop() {
        while (running) {
            try {
                Message message = messages.take();

                if (message != null) {
                    String emitterId = message.getEmitterId();
                    Emitter emitter = null;
                    EmitterType type = EmitterType.CLIENT;

                    emitter = clientEmitterMap.get(emitterId);
                    
                    if (emitter != null) {
                        if (emitter.getSink() != null) {
                        	
                            WebSocketMessage msg = (message.getBuffer() == null) ? textMessage(message.getMessage())
                                    : binaryMessage(message.getBuffer());
                            emitter.getSink().next(msg);
                            
                            if (type == EmitterType.CLIENT) {
                            	//ClientEmitter clientEmitter = (ClientEmitter)emitter;
                            	
                            	if("end".equalsIgnoreCase(message.getAlignmentKey())) {
                            		log.info("stop info.id:{} message: {}", emitter.getId(), message);
                            		//immediateSendCustomerClient(emitterId, JSON.toJSONString(Result.fail(ErrorDescEnum.session_close.getCode(),"",ErrorDescEnum.session_close.getMessage())));
                            		updateClientState(emitterId, EmitterState.CLOSEING);
                            	}
                            	//往大数据存日志
//                            	try {
//                            		JSONObject responeMessageObject = JSONObject.parseObject(message.getMessage());
//                            		if(responeMessageObject.containsKey("eof") && responeMessageObject.getJSONObject("result").containsKey("rec") && (0 == responeMessageObject.getInteger("eof") || 1 == responeMessageObject.getInteger("eof") || 2 == responeMessageObject.getInteger("eof"))) {
//                            				clientEmitter.getClientAckList().add(System.currentTimeMillis());
//                            				//BigDataLogger.logCloudvad(clientEmitter, responeMessageObject, message);
//                            		}
//
//                            	}catch(Exception e) {
//                            		log.info("=====Exception====message:{}",message.getMessage());
//                            		log.error("=====Exception====",e);
//                            	}
        		            }else {
//        		            	if(message.getBuffer() != null && (ServerType.CASR == emitter.getServerType() || ServerType.LASR == emitter.getServerType())) {
//        		            		log.info("=====send====requestId:{},size:{}",message.getMessage(), message.getBuffer().readableByteCount());
//        		            	}
        		            	log.error("emitter is not client. emitter: {}. message: {}", emitter, message);
        		            }
                        } else {
                            log.error("sink is null. emitter: {}. message: {}", emitter, message);
                            messages.add(message);
                        }
                    } else {
                        log.warn("Not found emitter: {}. message: {}", emitter, message);
                    }
                } else {
                    continue;
                }
            } catch (Exception e) {
            	log.error("IntegrationLoop failed",e);
                continue;
            }
        }
    }
    
    private void closeWebsocketSessionLoop() {
        while (running) {
        	if(clientEmitterMap != null && clientEmitterMap.size()>0) {
        		for(String sessionId: clientEmitterMap.keySet()) {
        			Emitter emitter= clientEmitterMap.get(sessionId);
        			if(emitter.getLastReceiveTime() > 0 ) {
        				if(System.currentTimeMillis()-emitter.getLastReceiveTime()>30000) {
            				log.info("======closeWebsocketSessionLoop time out closed===session id:{}",sessionId);
            				removeClientEmitter(sessionId);
            			}
        			}
        		}
        	}
        	try {
				Thread.sleep(30000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        }
    }
    
    
}
